# pip install redis pandas openpyxl

import redis
import queue
import pandas as pd
from threading import Thread
import time
import json
import logging.config

logging.config.fileConfig('./logging.ini')
logger = logging.getLogger('main')

# ========== 配置项 ==========
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
EXCEL_PATH = 'data_output.xlsx'
QUEUE_MAX_SIZE = 1000  # 队列最大容量
BATCH_SIZE = 100       # 批量写入阈值
# ===========================

# 初始化线程安全队列
data_queue = queue.Queue(maxsize=QUEUE_MAX_SIZE)

def redis_subscriber():
    """Redis数据订阅线程"""
    r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=False)
    logger.debug("connect to {}:{}".format(REDIS_HOST, REDIS_PORT))
    ps = r.pubsub()
    
    while True:
        # 订阅指定频道（可扩展多个）
        ps.subscribe('packet_queue')
    
        # 持续监听消息
        for message in ps.listen():
            logger.debug(type(message))
            '''
            if message['type'] == 'message':
                try:
                    # 解析JSON格式数据[3](@ref)
                    parsed_data = json.loads(message['data'])
                
                    # 添加时间戳
                    parsed_data['timestamp'] = time.strftime("%Y-%m-%d %H:%M:%S")
                
                    # 入队操作（阻塞直到有空位）
                    data_queue.put(parsed_data)
                except Exception as e:
                    print(f"数据解析失败: {e}")
'''
def excel_writer():
    """Excel写入线程"""
    df = pd.DataFrame(columns=['timestamp', 'device_id', 'value', 'status'])
    last_save_time = time.time()
    
    while True:
        batch_data = []
        
        # 批量获取数据
        while not data_queue.empty() and len(batch_data) < BATCH_SIZE:
            batch_data.append(data_queue.get())
        
        # 合并数据
        if batch_data:
            new_df = pd.DataFrame(batch_data)
            df = pd.concat([df, new_df], ignore_index=True)
        
        # 定时保存条件（30秒或达到批量阈值）
        if (time.time() - last_save_time > 30) or (len(df) >= BATCH_SIZE):
            try:
                # 写入Excel（追加模式）[10](@ref)
                with pd.ExcelWriter(EXCEL_PATH, engine='openpyxl', mode='a', if_sheet_exists='replace') as writer:
                    df.to_excel(writer, index=False, sheet_name='实时数据')
                print(f"成功写入{len(df)}条数据")
                df = df.iloc[0:0]  # 清空DataFrame
                last_save_time = time.time()
            except Exception as e:
                print(f"Excel写入失败: {e}")
        
        time.sleep(1)

if __name__ == "__main__":
    # 启动线程
    Thread(target=redis_subscriber, daemon=True).start()
    #Thread(target=excel_writer, daemon=True).start()
    
    # 主线程保持运行
    while True:
        time.sleep(3600)

